package com.cn.scenario.util.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.cn.scenario.util.mq.pojo.StrsqlResData;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.LocalDate;

/**
 * @Author Boolean
 * @Date 2023/2/21 18:24
 * @Version 1.0
 */
@Component
public class ResDataConsumer {

    /**
     * 消费者的组名
     */
    @Value("${rocketmq.conumerGroup}")
    private String consumerGroup;

    /**
     * NameServer 地址
     */
    @Value("${rocketmq.namesrvaddr}")
    private String namesrvAddr;


    //@PostConstruct
    public void consumer() {
        System.err.println("init defaultMQPushConsumer");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            //每次拉取间隔
            consumer.setPullInterval(1000);
            //每次拉取数量,注意：需要和mq的节点数相结合  *4 //http://www.360doc.com/content/20/0512/12/58006001_911755967.shtml
            consumer.setPullBatchSize(100);
            consumer.setMessageModel(MessageModel.BROADCASTING);
            consumer.subscribe("check-topic", "check-tag");
            System.out.println("------:"+System.currentTimeMillis());
            consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
                try {
                    for (MessageExt messageExt : list) {

                        System.err.println("消费消息: " + new String(messageExt.getBody()));//输出消息内容
                        StrsqlResData strsqlResData = JSON.parseObject(new String(messageExt.getBody()), StrsqlResData.class);
                        System.out.println("数据详情:"+strsqlResData.getId()+"--1---"+strsqlResData.getMsg());
                        //调用保存服务 Service
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

